// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jshin
#pragma once

#include <map>
#include <memory>
#include <stdexcept>
#include <string>
#include <unordered_map>
#include <vector>

#include <boost/asio/io_context.hpp>
#include <boost/asio/io_context_strand.hpp>
#include <boost/asio/post.hpp>
#include <zmqpp/zmqpp.hpp>
#include <fmt/format.h>

#include "coin2/base/config/app_config.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/feed/system.h"
#include "coin2/exchange/base/feed/update.h"
#include "coin2/exchange/base/feed_util/book_builder.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/base/symbology/product.h"
#include "coin2/exchange/di/constant.h"
#include "coin2/exchange/factory/feed/raw_processor.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "coin2/flow/topic_subscription.h"

namespace coin2::feed::relay {

using coin2::exchange::base::market::ExchangeType;
using coin2::exchange::base::market::MarketType;
using coin2::exchange::base::symbology::IProduct;
using coin2::exchange::feed::FeedSystemConfig;

// 1 FastFeedSubsystem = 1 mear
// so we can control root_dir, machine, etc per exchange.
class FeedRelaySubsystem : public coin2::exchange::base::feed::BaseLiveFeedSubsystem {
 public:
  FeedRelaySubsystem(
      const std::string& system_name,
      const MarketExchangeApiRecipe& mear,
      const coin2::exchange::di::ConstantInjection& ci,
      const FeedSystemConfig& config,
      exchange::base::executor::LiveAsyncExecutor* lass,
      exchange::base::feed::IFeedSubscriber* callback)
      : coin2::exchange::base::feed::BaseLiveFeedSubsystem(system_name, mear.mea, lass),
        mear_(mear),
        mea_string_(mear.mea.SnakeString()),
        // ci_(ci),
        use_arb_(config.websocket().arbitration()),
        ioc_(lass->MainIoc()),
        ioc_lifetime_guard_(ioc_->get_executor()),
        zmq_url_(config.relay_server_zmq_url()),
        processor_(coin2::exchange::factory::feed::NewRawProcessor(
            ci,
            system_name,
            mear.mea,
            config.raw_parser(),
            config.recipe())) {
    auto mea_string = mear.mea.String();
    auto mear_string = mear.String();

    const auto& group_config = ci.GetSymbolGroups();

    auto products = ProductRequestsFromConfig(mea_string, config.products());
    for (auto& prod : products) {
      auto symbol = prod->feed_native_symbol();
      processor_->AddProduct(*prod, callback);
      zmq_topics_.push_back(mear_string + ":" + prod->relative_norm());

      std::string relay_topic = fmt::format("{}:{}", mear.String(), prod->relative_norm());
      char group = group_config.GetGroupForRelativeNorm(mear.mea, prod->relative_norm());
      relay_topic_to_lass_topic_[relay_topic] = fmt::format("{}_{}1", mea_string_, group);
    }

    if (use_arb_) {
      // constexpr int num_worker = 1;
      // topic_register_.RegisterLiveLongTopicArb(
      //     mear.mea, processor_.get(), lass->distributor(),
      //     ci_.GetArbitrationConfigs(), num_worker);
    } else {
      topic_register_.RegisterLiveShortTopic(mear.mea, processor_.get(), lass->distributor());
    }

    LOG(INFO) << "< " << config.products().DebugString() << " >";
    LOG(INFO) << "[Feed Relay System]";
    LOG(INFO) << "              mea: " << mea_.String();
    LOG(INFO) << "           recipe: " << mear_.recipe;
    for (const auto& topic : zmq_topics_) {
      LOG(INFO) << "            topic: " << topic;
    }
  }

  std::vector<IProduct*> GetProducts() const override {
    std::vector<IProduct*> out;
    for (const auto& [k, v] : products_) {
      (void)k;
      out.push_back(v.get());
    }
    return out;
  }

  int cnt = 0;
  void StartAsync() override {
    if (is_running_) {
      throw std::runtime_error("StartAsync(): zmq PubSub already running");
    }
    is_running_ = true;

    thread_ = std::thread(&FeedRelaySubsystem::Loop, this);
  }

  void Loop() {
    zmqpp::context context;
    zmqpp::socket socket(context, zmqpp::socket_type::sub);
    socket.connect(zmq_url_);
    // socket.subscribe("Futures.Okex.v3.realtime_light_bbo:BTC-USD.QUARTER");

    for (const auto& topic : zmq_topics_) {
      socket.subscribe(topic);
    }

    while (is_running_) {
      zmqpp::message message;
      socket.receive(message);

      // // Read as a string
      std::string relay_topic;  // mear:symbol
      std::string rt;     // int64
      std::string msg;    // string
      std::string lass_topic;  // lass_topic

      int64_t ts = coin2::exchange::base::executor::TopicRecordProcessor::GetTimestamp();

      message >> relay_topic;
      message >> rt;
      message >> msg;

      if (use_arb_) {
        // "spot_okex_v3_a1"
        lass_topic = relay_topic_to_lass_topic_[relay_topic];
      } else {
        // "spot_okex_v3"
        lass_topic = mea_string_;
      }

      boost::asio::post(
          *ioc_,
          [this, lass_topic, ts, msg] {  // capture msg by value, since this closure can run later
            auto packet = std::shared_ptr<::coin2::exchange::base::executor::TopicRecordPacket>(
                new ::coin2::exchange::base::executor::TopicRecordPacket{
                    lass_topic,
                    0,
                    ts,
                    0,
                    0,
                    0,
                    0,
                    0,
                    0,
                    msg});
            live_executor_->injector().push(packet);
          });
    }
  }

  void StopAsync() override {
    if (!is_running_) {
      throw std::runtime_error("StopAsync(): zmq PubSub was not running");
    }
    is_running_ = false;
  }

 private:
  const MarketExchangeApiRecipe mear_;
  std::string mea_string_;
  // const coin2::exchange::di::ConstantInjection& ci_;
  const bool use_arb_;

  boost::asio::io_context* ioc_;
  boost::asio::executor_work_guard<boost::asio::io_context::executor_type>
      ioc_lifetime_guard_;  // prevent ioc to finish

  std::string zmq_url_;
  std::unique_ptr<coin2::exchange::base::feed::FeedParsingProcessor> processor_;
  std::vector<std::string> zmq_topics_;

  bool is_running_ = false;
  std::thread thread_;
  std::map<std::string, std::unique_ptr<IProduct>> products_;

  coin2::exchange::base::feed::TopicRegister topic_register_;
  std::unordered_map<std::string, std::string> relay_topic_to_lass_topic_;
};

}  // namespace coin2::feed::relay
